package com.shujia.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo01WordCount {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName("streaming")
      .setMaster("local[2]")

    /**
      * 创建steaming上下文对象，指定batch的间隔时间，多久计算一次
      */
   val ssc: StreamingContext = new StreamingContext(conf,Durations.seconds(5))

    //设置checkpoint
    /**
      * Checkpoint 的产生就是为了相对而言更加可靠的持久化数据，
      * 在 Checkpoint 可以指定把数据放在本地并且是多副本的方式，
      * 在正常生产环境下通常放在 HDFS 上，借助HDFS 高可靠的特征来实现更可靠的数据持久化。
      *
      */
    ssc.checkpoint("data/checkpoint ")

    /**
      * ReceiverInputDStream：被动接受数据，将接受过来的数据放在内存和磁盘上
      * 接受数据会一直占用资源
      *
      *
      * nc -lk 8888
      * netstat -anp | grep 8888
      * yum install nc
      */
    //读取数据
    val linesDs: ReceiverInputDStream[String] = ssc.socketTextStream("master",8888)

    //切分数据
    val wordsDS: DStream[String] = linesDs.flatMap(_.split(","))

    //转化为kv格式
    val kvDS: DStream[(String, Int)] = wordsDS.map((_,1))

    //计算
//    val countDS: DStream[(String, Int)] = kvDS.reduceByKey(_+_)

    def updataFun(seq: Seq[Int],option: Option[Int]):Option[Int] = {
      //计算当前batch单词的数量
      val currCount: Int = seq.sum

      //获取之前单词的数量
      val lastCount: Int = option.getOrElse(0)

      //返回最新单词的数量
      Some(currCount + lastCount)
    }

    /**
      *
      * 有状态算子
      *
      * updateStateByKey:每一个计算更新每一个key的状态（单词的数量）
      *
      * 需要设置checkpoint路径，用于保存计算过程中的状态
      */

    val countDS: DStream[(String, Int)] = kvDS.updateStateByKey(updataFun)
    //输出数据
    countDS.print()

    //启动spark steaming
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }
}
